package com.bw.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

//{"database":"tms01","data":{},"type":"insert","table":"user_info"}
public class HbaseSinkFunction extends RichSinkFunction<String> {
    Connection conn = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop102:2181");
        conn.setAutoCommit(true);
    }
    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        //名称前缀：dim_，表列簇名称：info
        //手动建表：create table TMS.dim_user_info (id varchar primary key,info.data varchar)
        //手动建表：create table TMS.dim_base_dic (id varchar primary key,info.data varchar)
        //手动建表：create table TMS.dim_base_region_info (id varchar primary key,info.data varchar)
        //手动建表：create table TMS.dim_express_courier (id varchar primary key,info.data varchar)

        JSONObject js = JSON.parseObject(value);
        String id = js.getJSONObject("data").getString("id");
        String data = js.getString("data");
        String tableName = "dim_"+js.getString("table");

        // id,data
        PreparedStatement ps = conn.prepareStatement("upsert into TMS." + tableName + " values (?,?)");
        ps.setString(1,id);
        ps.setString(2,data);
        ps.execute();

        ps.close();
    }
}
